﻿using System;
using System.Collections.Generic;
using System.Text;
using RabbitMQ.Client;

namespace Having.Site.MessageQueue
{
    /// <summary>
    /// RabbitMQ
    /// </summary>
    public class RabbitMessageQueue : IMessageQueue
    {
        private ConnectionFactory rabbitMqFactory;

        /// <summary>
        /// 链接配置  
        /// </summary>
        public MQConfig ConnConfig
        {
            get; set;
        }
        private ExchangeType EType { set; get; }

        /// <summary>
        /// 初始化
        /// </summary>
        /// <param name="_connstring">配置信息</param>
        /// <param name="_etype">广播方式</param>
        public RabbitMessageQueue(MQConfig _connstring, ExchangeType _etype= ExchangeType.fanout)
        {
            this.ConnConfig = _connstring;
            this.EType = _etype;
        }
        /// <summary>
        /// 打开链接
        /// </summary>
        public void Open()
        {
            if (rabbitMqFactory == null)
            {
                try
                {
                    rabbitMqFactory = new ConnectionFactory()
                    {
                        HostName = ConnConfig.Host,
                        Port = ConnConfig.Port,  //5672 是针对一般而言的连接, 5671 是针对 TLS 连接的 
                        Password = ConnConfig.Password,
                        UserName = ConnConfig.UserName

                    };
                }
                catch
                {
                    //无法创建rabbit链接
                }
            }
        }

        /// <summary>
        /// 获得数量
        /// </summary>
        /// <returns></returns>
        public int Count()
        {
            uint c = 0;
            using (var conn = rabbitMqFactory.CreateConnection())
            {
                using (var channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(ConnConfig.Exchange, EType.ToString(), durable: true, autoDelete: false, arguments: null);
                    c = channel.MessageCount(ConnConfig.QueueName);
                }
            }
            return (int)c;
        }
        /// <summary>
        /// 释放
        /// </summary>
        public void Dispose()
        {
            if (rabbitMqFactory != null)
            {
                rabbitMqFactory = null;
            }
            GC.SuppressFinalize(this);
        }


        /// <summary>
        /// 清空队列
        /// </summary>
        public void Purge()
        {
            Open();
            using (var conn = rabbitMqFactory.CreateConnection())
            {
                using (var channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(ConnConfig.Exchange, EType.ToString(), durable: true, autoDelete: false, arguments: null);
                    channel.QueuePurge(ConnConfig.QueueName);
                }
            }
        }
        /// <summary>
        /// 接收到的消息对象(不断循环读取，该方法必须用线程)
        /// </summary>
        /// <param name="mi"></param>
        public void Receive( Action<string> mi )
        {
            Open();
            var conn = rabbitMqFactory.CreateConnection();
            var channel = conn.CreateModel();
            channel.ExchangeDeclare(ConnConfig.Exchange, EType.ToString(), durable: true, autoDelete: false, arguments: null);
            channel.QueueDeclare(ConnConfig.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
            if (EType != ExchangeType.fanout)
            {
                var sp = ConnConfig.RoutingKey.Split(new string[] { "," }, StringSplitOptions.RemoveEmptyEntries);
                foreach (var item in sp)
                {
                    channel.QueueBind(ConnConfig.QueueName, ConnConfig.Exchange, routingKey: item);
                }
            }
            else
            {
                channel.QueueBind(ConnConfig.QueueName, ConnConfig.Exchange, routingKey: "");
            }
            channel.BasicQos(0, 1, false);
            var consumer = new RabbitMQ.Client.Events.EventingBasicConsumer(channel);

            consumer.Received += (model, ea) =>
            {
                var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                channel.BasicAck(ea.DeliveryTag, true);
                mi(message);

            };
            channel.BasicConsume(queue: ConnConfig.QueueName, autoAck: false, consumer: consumer);
            //while (true)
            //{
            //    var rs = channel.BasicGet(QueueName, false);
            //    if (rs != null)
            //    {
            //        var ms = rs.Body;
            //        channel.BasicAck(rs.DeliveryTag, false);
            //        var s = Encoding.UTF8.GetString(ms.ToArray());
            //        mi(s);
            //    }
            //}

        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="message"></param>
        public void SendMessage(List<string> message)
        {
            Open();
            using (var conn = rabbitMqFactory.CreateConnection())
            {
                using (var channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(ConnConfig.Exchange, EType.ToString(), durable: true, autoDelete: false, arguments: null);
                    channel.QueueDeclare(ConnConfig.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
                    channel.QueueBind(ConnConfig.QueueName, ConnConfig.Exchange, routingKey: (EType == ExchangeType.fanout ? "" : ConnConfig.RoutingKey));
                    foreach (var str in message)
                    {
                        byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes(str);
                        channel.BasicPublish(ConnConfig.Exchange, ConnConfig.QueueName, null, messageBodyBytes);
                    }
                }
            }
        }
        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="message"></param>
        public void SendMessage(string message)
        {
            Open();
            using (var conn = rabbitMqFactory.CreateConnection())
            {
                using (var channel = conn.CreateModel())
                {
                    try
                    {
                        channel.ExchangeDeclare(ConnConfig.Exchange, EType.ToString(), durable: true, autoDelete: false, arguments: null);
                        channel.QueueDeclare(ConnConfig.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
                        channel.QueueBind(ConnConfig.QueueName, ConnConfig.Exchange, routingKey: (EType == ExchangeType.fanout ? "" : ConnConfig.RoutingKey));
                        byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish(ConnConfig.Exchange, ConnConfig.QueueName, null, messageBodyBytes);
                    }
                    catch
                    { }
                }
            }
        }

        /// <summary>
        /// 关闭链接并释放资源
        /// </summary>
        public void Close()
        {
            Dispose();
        }
    }
}
